package com.actor;

import java.util.HashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author Dmitry
 */
public class QueueProcessor {
    private final LinkedBlockingQueue<MessageWrapper> queue = new LinkedBlockingQueue<>();
    private final HashMap<Actor, Actor> actors = new HashMap<>();
    private Thread thread;

    public QueueProcessor() {
    }

    public void start() {
        thread = new Thread() {
            @Override
            public void run() {

                while (!isInterrupted()) {
                    try {
                        MessageWrapper message = queue.poll(1000, TimeUnit.MILLISECONDS);
                        if (message != null) {
                            processMessage(message);
                        }
                    } catch (InterruptedException ex) {
                        //ignore
                    }
                }
            }
        };
        thread.setDaemon(false);
        thread.start();
    }

    public void stop() {
        thread.interrupt();
        queue.clear();
        actors.clear();
    }

    private void processMessage(MessageWrapper message) {
        try {
            Actor actor = message.getReceiver();
            if (actor != null) {
                ActorRef sender = message.getSender() == null ? actor.getActorSystem().getNullActor() : message.getSender();
                if (message instanceof MessageWrapperFutureResponse) {
                    MessageWrapperFutureResponse mwf = (MessageWrapperFutureResponse) message;
                    Future future = (Future) actor.getFutures().get(mwf.getFutureTag());
                    if (future != null) {
                        actor.getFutures().remove(mwf.getFutureTag());
                        if (mwf.getError() != null) {
                            future.onError(mwf.getValue(), mwf.getError(), sender);
                        } else {
                            future.onSuccess(mwf.getValue(), sender);
                        }

                        future.onComplete(mwf.getValue(), sender);

                    }
                } else {
                    try {
                        actor.receive(message.getValue(), sender);
                    } catch (Throwable thr) {

                        sender.onError(thr, message.getValue(), sender);
                    }
                }
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public void addMessage(MessageWrapper message) {
        queue.offer(message);
    }

    public int size() {
        return actors.size();
    }

    public void addActor(Actor actor) {
        actors.put(actor, actor);
        actor.setActorQueue(this);
    }

    public boolean hasActor(Actor actor) {
        return actors.containsKey(actor);
    }

    public void removeActor(Actor actor) {
        actors.remove(actor);
        actor.setActorQueue(null);
    }
}
